package com.im.service.message.mq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.im.common.constant.Constants;
import com.im.common.enums.command.MessageCommand;
import com.im.common.model.message.MessageContent;
import com.im.common.model.message.MessageReadedContent;
import com.im.common.model.message.MessageReciveAckContent;
import com.im.common.model.message.RecallMessageContent;
import com.im.service.message.service.MessageSyncService;
import com.im.service.message.service.P2PMessageService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import java.util.Map;
import java.util.Objects;

/**
 * 聊天操作接收类
 */
@Slf4j
@Component
public class ChatOperateReceiver {

    // 注入 点对点消息服务类 和 消息同步服务类
    @Autowired
    private P2PMessageService p2PMessageService;
    @Autowired
    private MessageSyncService messageSyncService;

    /**
     * 监听 RabbitMQ 队列：
     *      bindings：表示队列和交换机之间的绑定关系。它是一个数组，可以包含多个@QueueBinding注解。
     *      value：表示队列或交换机的名称。在这里，它表示IM2消息服务队列和同名的交换机。
     *      durable：表示队列或交换机是否持久化。在这里，它表示IM2消息服务队列和同名的交换机都是持久化的。
     *      exchange：表示交换机的相关属性。在这里，它表示IM2消息服务队列和同名的交换机是同一个交换机。
     *      concurrency：表示消费者的并发数量。在这里，它表示只有一个消费者会处理IM2消息服务队列中的消息。
     */
    @RabbitListener(
            bindings = @QueueBinding( // 用于将队列和交换机绑定在一起
                    value = @Queue(value = Constants.RabbitConstants.Im2MessageService, durable = "true"), // 用于定义一个队列
                    exchange = @Exchange(value = Constants.RabbitConstants.Im2MessageService, durable = "true") // 用于定义一个交换机
            ), concurrency = "1"
    )
    public void onChatMessage(@Payload Message message, @Headers Map<String, Object> headers, Channel channel) throws Exception {
        // 将消息体转换为字符串
        String msg = new String(message.getBody(), "utf-8");
        log.info("CHAT MSG FORM QUEUE ::: {}", msg);
        // 获取消息的 deliveryTag
        Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
        try {
            // 解析消息体
            JSONObject jsonObject = JSON.parseObject(msg);
            Integer command = jsonObject.getInteger("command");
            if (command.equals(MessageCommand.MSG_P2P.getCommand())) {
                // 处理 P2P 消息
                MessageContent messageContent = jsonObject.toJavaObject(MessageContent.class);
                p2PMessageService.process(messageContent);
            } else if (command.equals(MessageCommand.MSG_RECIVE_ACK.getCommand())) {
                // 处理消息接收确认
                MessageReciveAckContent messageContent = jsonObject.toJavaObject(MessageReciveAckContent.class);
                messageSyncService.receiveMark(messageContent);
            } else if (command.equals(MessageCommand.MSG_READED.getCommand())) {
                // 处理消息已读确认
                MessageReadedContent messageContent = jsonObject.toJavaObject(MessageReadedContent.class);
                messageSyncService.readMark(messageContent);
            } else if (Objects.equals(command, MessageCommand.MSG_RECALL.getCommand())) {
                // 处理消息撤回
                RecallMessageContent messageContent = JSON.parseObject(msg, new TypeReference<RecallMessageContent>() {
                }.getType());
                messageSyncService.recallMessage(messageContent);
            }
            // 手动 ACK 消息
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            log.error("处理消息出现异常：{}", e.getMessage());
            log.error("RMQ_CHAT_TRAN_ERROR", e);
            log.error("NACK_MSG:{}", msg);
            // 第一个 false 表示不批量拒绝，第二个 false 表示不重回队列
            channel.basicNack(deliveryTag, false, false);
        }
    }

}

